package com.codingapi.websocket.core.handler;


import com.codingapi.websocket.core.client.DefaultWebSocketSession;
import com.codingapi.websocket.core.client.WebSocketSession;
import com.codingapi.websocket.core.handler.netty.AbstractInboundHandler;
import com.codingapi.websocket.core.handler.netty.AbstractOutboundHandler;
import com.codingapi.websocket.core.resolver.UpgradeResolver;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpServerKeepAliveHandler;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.websocketx.*;

import java.net.URI;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static java.util.concurrent.Executors.newScheduledThreadPool;


/**
 * @author modificial
 * @date 2018/7/19
 * @company codingApi
 * @description 产生websocket处理器工厂
 */
public class WebSocketChannelHandlerFactory {

    /**
     * 全局websocket连接对象存储
     */

    private static Map<String, WebSocketSession> webSocketSessions = new ConcurrentHashMap<>();

    public ChannelInboundHandlerAdapter newWSInboundHandler() {
        return new WSInboundHandler();
    }

    public ChannelOutboundHandlerAdapter newWSOutboundHandler() {
        return new WSOutboundHandler();
    }

    /**
     * 消息处理
     */
    private static class WSInboundHandler extends AbstractInboundHandler {

        /**
         *  websocket 升级请求处理器
          */
        private UpgradeResolver upgradeResolver = new UpgradeResolver();
        /**
         *  默认请求处理器
         */
        private WebSocketHandler webSocketHandler;

        public WSInboundHandler() {
            this.webSocketHandler = new SimpleWebSocketHandler();
        }

        public WSInboundHandler(WebSocketHandler webSocketHandler) {
            this.webSocketHandler = webSocketHandler;
        }


        @Override
        public void doChannelRead(String id, ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof FullHttpRequest) {
                //处理http请求
                FullHttpRequest request = (FullHttpRequest) msg;
                String uri = request.uri();
                QueryStringDecoder uriDecoder = new QueryStringDecoder(uri);
                Map<String, List<String>> parameters = uriDecoder.parameters();
                String path = uriDecoder.path();
                if (path.startsWith("//")) {
                    path = path.substring(1);
                }

                DefaultWebSocketSession webSocketSession = new DefaultWebSocketSession();
                webSocketSession.setParameters(parameters);
                webSocketSession.setHandshakeHeaders(request.headers());
                webSocketSession.setId(id);
                webSocketSession.setUri(new URI(path));
                webSocketSession.setChannelHandlerContext(ctx);
                //获取请求处理器
                WebSocketHandler socketHandler = this.getSocketHandler();
                if (socketHandler == null) {
                    webSocketSession.setWebSocketHandler(this.webSocketHandler);
                } else {
                    webSocketSession.setWebSocketHandler(new SimpleWebSocketHandler());
                }
                socketHandler = webSocketSession.getWebSocketHandler();

                //upgrade之前的回调 执行websocket前拦截，作参数的处理
                try {
                    socketHandler.beforeConnectionUpgraded(webSocketSession);
                } catch (Exception e) {
                    //处理异常 没有具体映射的请求处理器
                    socketHandler.handleTransportError(webSocketSession, e);
                    return;
                }

                WebSocketServerHandshaker handshaker = null;
                // upgrade 与 websocket 握手过程
                if ((handshaker = upgradeResolver.handleRequest(ctx, request)) != null) {
                    webSocketSession.setWebSocketServerHandshaker(handshaker);
                    //完成后调用
                    socketHandler.afterConnectionEstablished(webSocketSession);
                    webSocketSessions.put(id, webSocketSession);
                } else {
                    socketHandler.handleTransportError(webSocketSession, new RuntimeException("upgrade failed.!"));
                }

            } else if (msg instanceof WebSocketFrame) {
                //处理websocket请求
                WebSocketSession webSocketSession = webSocketSessions.get(id);
                WebSocketHandler webSocketHandler = null;
                if (webSocketSession == null) {
                    webSocketHandler = this.webSocketHandler;
                } else if (msg instanceof CloseWebSocketFrame) {
                    //浏览器关闭后， WebSocket 对象会发送close ，此处进行处理
                    WebSocketSession wsSession = webSocketSessions.remove(id);
                    if (wsSession != null) {
                        wsSession.close();
                        wsSession.getWebSocketHandler().afterConnectionClosed(webSocketSession);
                    }
                    KeepAliveHandlerAdapter.receivePongMessage(id);
                } else {
                    webSocketHandler = webSocketSession.getWebSocketHandler();
                    webSocketHandler.handleMessage(webSocketSession, (WebSocketFrame) msg);
                }
                KeepAliveHandlerAdapter.receivePongMessage(id);
            } else {
                throw new RuntimeException("无法处理的请求");
            }
        }


        @Override
        public void doExceptionCaught(String channelId, ChannelHandlerContext ctx, Throwable cause) throws Exception {
            WebSocketSession webSocketSession = webSocketSessions.get(channelId);
            if (webSocketSession != null) {
                webSocketSession.getWebSocketHandler().handleTransportError(webSocketSession, cause);
            }
        }

        public WebSocketHandler getWebSocketHandler() {
            return webSocketHandler;
        }

        public void setWebSocketHandler(WebSocketHandler webSocketHandler) {
            this.webSocketHandler = webSocketHandler;
        }

        public UpgradeResolver getUpgradeResolver() {
            return upgradeResolver;
        }

        public void setUpgradeResolver(UpgradeResolver upgradeResolver) {
            this.upgradeResolver = upgradeResolver;
        }

        private WebSocketHandler getSocketHandler() {
            if (webSocketHandler != null) {
                return webSocketHandler;
            }
            return null;
        }
    }

    /**
     * 关闭连接
     */
    private static class WSOutboundHandler extends AbstractOutboundHandler {

        @Override
        public void doDisconnect(String channelId, ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        }

        @Override
        public void doClose(String channelId, ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            WebSocketSession webSocketSession = webSocketSessions.remove(channelId);
            if (webSocketSession != null) {
                webSocketSession.close();
                webSocketSession.getWebSocketHandler().afterConnectionClosed(webSocketSession);
            }
            KeepAliveHandlerAdapter.receivePongMessage(channelId);
        }

    }


    /**
     * 服务端保活清除过期或者失联连接 处理器
     * 微信浏览器打开页面关闭后(或者另外的场景)不会触发close消息 ，所以又做了一层心跳机制
     * <p>
     * <p>
     * 策略是通过定时器
     * 1.遍历客户端发送ping消息 ，然后在缓存中 pingPongMap  标记 1已发送 (或者n 表示发送次数)
     * 2.客户端接收ping 消息会返回pong 消息 (websocket协议标准 见RFC6455)
     * 3.在接收端 doHandleRequest() 方法中(接收到非close frame )说明客户端有反馈，表示存活，删除 pingPongMap
     * 4.接着下一个周期定时器去遍历 pingPongMap 查出已经发送ping 无响应超过 MAX_RE_PING 次数
     * 5.删除超过 MAX_RE_PING 无响应的客户端
     * 6.每次接收到客户端的消息，都会调用 doHandleRequest() 方法清除 pingPongMap 表示客户端存活
     */
    private static class KeepAliveHandlerAdapter extends HttpServerKeepAliveHandler {

        /**
         * 轮训时间 检测过期连接 定时器定时时间
         */
        private final static int SCHEDULE_SECONDS = 5;
        private static final ThreadLocal<ScheduledExecutorService> scheduleService = ThreadLocal.withInitial(() -> newScheduledThreadPool(10));

        /**
         * 心跳  已发送次数
         */
        private static ConcurrentHashMap<String, Integer> pingPongChannelsMap = new ConcurrentHashMap<>();

        static {
            scheduleService.get().scheduleAtFixedRate(() -> {
                try {
                    //定时发送心跳
                        sendPingMessageToAll();
                } catch (Exception ignored) {
                }
            }, 1L, SCHEDULE_SECONDS, TimeUnit.SECONDS);
        }

        /**
         * 给所有客户端发送ping 消息
         */
        private static void sendPingMessageToAll() {
            Collection<WebSocketSession> sessions = WebSocketChannelHandlerFactory.webSocketSessions.values();
            if (Objects.nonNull(sessions)) {
                for (WebSocketSession socketSession : sessions) {
                    TextWebSocketFrame ping=new TextWebSocketFrame("ping");
                    socketSession.sendMessage(ping);
                }
            }
        }

        /**
         * 收到消息表示存活
         *
         * @param id
         */
        private static void receivePongMessage(String id) {
            KeepAliveHandlerAdapter.pingPongChannelsMap.remove(id);
        }

    }

}
